/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import org.sss.client.DataGetter;
import org.sss.client.DataPutter;
import org.sss.client.JobEngine;
import org.sss.client.SssClient;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.Mapper;
import org.sss.mapreduce.Output;
import org.sss.mapreduce.Tuple;
import org.sss.mapreduce.datatype.PackableBytes;
import org.sss.mapreduce.datatype.PackableLong;
import org.sss.mapreduce.datatype.PackableString;

public class TupleReaderExample {

  private static final String CONF_KEY = "tuplereader.tuples";

  public static class MyMapper extends Mapper {

    long result = 0;

    @Override
    public void configure(Context context) throws Exception {
      GroupID gid = GroupID.createFromString(context.getConfiguration().get(CONF_KEY));
      result = 0;
      for (Tuple<PackableLong, PackableLong> t: context.getTupleReader(gid, PackableLong.class, PackableLong.class)) {
          assert t.getKey().equals(t.getValue());
          result += t.getValue().get();
      }
    }

    public void map(Context context,
                    PackableBytes key, PackableBytes value,
                    Output<PackableString, PackableLong> output) throws Exception {
      // Ignore key&value.
      output.write(new PackableString("dummyKey"), new PackableLong(result));
    }
  }

  public static void main(String[] args) throws Exception {
    SssClient client = new SssClient(args);

    String[] others = client.getArgs();
    if (others.length < 1) {
      System.err.println("input group id is not specified.");
      System.exit(1);
    }
    long n = Long.parseLong(others[0]);

    // Put
    DataPutter<PackableLong, PackableLong> putter = DataPutter.create(client, PackableLong.class, PackableLong.class);
    try {
      for (long i = 0;i < n;++i) {
        putter.append(new PackableLong(i), new PackableLong(i));
      }
    } finally {
      putter.close();
    }
    GroupID input = putter.getOutputGroupID();

    DataPutter<PackableBytes, PackableBytes> dummyPutter = DataPutter.create(client, PackableBytes.class, PackableBytes.class);
    try {
      dummyPutter.append(new PackableBytes(new byte[0]), new PackableBytes(new byte[0]));
    } finally {
      dummyPutter.close();
    }
    GroupID dummyInput = dummyPutter.getOutputGroupID();

    // Job
    JobEngine engine = new JobEngine(client);
    GroupID output;
    try {
      output = GroupID.createRandom(engine);
      engine.getConfiguration().set(CONF_KEY, input.toString());

      engine.getJobBuilder(MyMapper.class)
        .addInput(dummyInput)
        .addOutput(output).build();
      engine.exec();
    } finally {
      engine.dispose();
    }

    long result = 0;
    for (long i = 0;i < n;++i) {
      result += i;
    }

    // Job
    DataGetter<PackableString, PackableLong> getter =
      DataGetter.create(client, PackableString.class, PackableLong.class, output);
    boolean error = false;
    try {
      long size = 0;
      for (Tuple<PackableString, PackableLong> kv: getter) {
        size++;
        if (!kv.getKey().get().equals("dummyKey")) {
          System.err.println("Unexpected key: " + kv.getKey().get());
          error = true;
        }
        if (kv.getValue().get() != result) {
          System.err.println("Unexpected value: " + kv.getValue().get());
          error = true;
        }
      }
      if (size != 1) {
          System.err.println("Unexpected tuple group size.");
          error = true;
      }
    } finally {
      getter.close();
    }
    if (error) {
      System.exit(1);
    }
  }
}
